#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <errno.h>
#include <sys/uio.h>
#include "queue.h"
#include "string.h"

/* notre structure queue */

struct queue { 

  void **buffer;
  size_t *bufsize; /* espace des différents éléments */

  int in; /* indice ou l'on place un element */
  int out; /* indice ou l'on doit remove un element */
  int nbr; /* nombre d'éléments dans la file */
  int size; /* taille max du buffer (alloue queue_new) */

  /* mutex */
  pthread_mutex_t mtx;

  /* pthread cond */
  pthread_cond_t empty;
  pthread_cond_t full;
};


/* lorsque la file est vide met les pointeurs in et out a 0 
   evite ainsi de doubler la taille du buffer lorsque la file est vide 
*/
int args_reset(struct queue *queue)
{
  if(queue->nbr == 0 && queue->in != 0 && queue->out != 0 )
    {
      queue->in = queue->out = 0;
      return 0;
    }

  return -1;

}

struct queue *queue_new(size_t size)
{
  struct queue *queue;
  int err_sys;

  queue = malloc(sizeof(struct queue));
  if(queue == NULL){
    perror("malloc queue error");
    err_sys = 1;
    /* pthread_exit(&err_sys);*/
    exit(1);
  }

  queue->in = 0;
  queue->out = 0;
  queue->nbr = 0;
  queue->size = size;
  
  queue->buffer = (void **) calloc(size,sizeof(void*));
  if(queue->buffer == NULL){
    perror("calloc buffer error");
    exit(1);
  }

  queue->bufsize = (size_t *) calloc(size,sizeof(size_t));
  if(queue->bufsize == NULL){
    perror("calloc size buffer error");
    exit(1);
  }

  /** initialisation mutex & variable conditions **/

  err_sys = pthread_mutex_init(&queue->mtx,NULL);
  if(err_sys > 0){
    perror("mutex init error");
    exit(1);
  }
  
  err_sys = pthread_cond_init(&queue->empty,NULL);
  if(err_sys > 0){
    perror("condition empty init error");
    exit(1);
  }

  err_sys = pthread_cond_init(&queue->full,NULL);
  if(err_sys > 0){
    perror("condition full init error");
    exit(1);
  }

  return queue;
}


void queue_destroy(struct queue *queue)
{
  int err_sys;

  /* on supprime la mutex et les variables conditions */
  err_sys =  pthread_mutex_destroy(&queue->mtx);
  if(err_sys > 0){
    perror("mutex destroy error");
    exit(1);
  }
  
  err_sys = pthread_cond_destroy(&queue->empty);
   if(err_sys > 0){
    perror("condition empty destroy error");
    exit(1);
  }

  err_sys = pthread_cond_destroy(&queue->full);
  if(err_sys > 0){
    perror("condition full destroy error");
    exit(1);
  }
  
  /* on free le buffer et ensuite la queue */
  free(queue->buffer);
  queue->buffer = NULL;
  
  free(queue);
 
}

/****

realloc sur temp, puis si seulement non null alors faire realloc btemp

 ****/


int enqueue_block(struct queue *queue, const void *buf, size_t size)
{  
  int err_thr, retval;
  void *tmp;
  
  err_thr = pthread_mutex_lock(&(queue->mtx));
  if( err_thr > 0){
    perror("mutex lock error");
    retval = 1;
    pthread_exit(&retval);
  }   
  
  if(opt == 0) {

    while( queue->nbr == queue->size) 
      {
	err_thr = pthread_cond_wait(&(queue->full), &(queue->mtx));
	if( err_thr > 0) {
	  perror("cond full wait error");
	  pthread_exit(NULL);
	}	
      
      }
  } 
  
  else if( opt == 2 ){
    //args_reset(queue);

    if (queue->in == queue->size ) {
      size_t *btemp;
      void **temp;

      temp = realloc(queue->buffer,
		     (size_t)2*queue->size*sizeof(void*));
    
      while( temp == NULL) 
	{
	  err_thr = pthread_cond_wait(&(queue->full), &(queue->mtx));
	  if( err_thr > 0) {
	    perror("double cond full wait error");
	    pthread_exit(NULL);
	  }
	  temp = realloc(queue->buffer,
			 (size_t)2*queue->size*sizeof(void*));

	}
    
      btemp = realloc(queue->bufsize,
		      (size_t)2*queue->size*sizeof(size_t));

      while( btemp == NULL) 
	{
	  err_thr = pthread_cond_wait(&(queue->full), &(queue->mtx));
	  if( err_thr > 0) {
	    perror("double cond full wait error");
	    pthread_exit(NULL);
	  }
	  btemp = realloc(queue->bufsize,
			  (size_t)2*queue->size*sizeof(size_t));

	}
      /* si le realloc marche, on pointe le buffer et on double size */
      
      queue->buffer = temp;
      queue->bufsize = btemp;
      queue->size = queue->size*2; 
    }
  }
  
  tmp  = (void*) malloc(size);
  while( tmp == NULL) {
    err_thr = pthread_cond_wait(&(queue->full),&(queue->mtx));
    if(err_thr > 0)
      {
	perror("enqueue cond full wait error");
	pthread_exit(NULL);
      }
    tmp = (void*) malloc(size);
  }
  
  queue->buffer[queue->in%queue->size] = tmp;
  queue->buffer[queue->in%queue->size] 
    = memmove(queue->buffer[queue->in%queue->size],buf,size);

  if(queue->buffer[queue->in%queue->size] == NULL){
    perror("memmove error enqueue");
    retval = 2;
    pthread_exit(&retval);
  }

  queue->bufsize[queue->in%queue->size] = size;
  queue->in++;
  queue->nbr++;

  /*
  int *r = (int*)buf;
  printf("enqueue %d\n",(int)*r);
  */

  err_thr =  pthread_mutex_unlock(&(queue->mtx));
  if( err_thr > 0){
    perror("mutex unlock error");
    retval = 1;
    pthread_exit(&retval);
  } 
  
  err_thr = pthread_cond_signal(&(queue->empty));
  if( err_thr > 0){
    perror("condition signal error");
    retval = 1;
    pthread_exit(&retval);
  } 

  return 0;
}

int enqueue_nblock(struct queue *queue, const void *buf, size_t size)
{
  int err_thr, retval;
  void *tmp;

  err_thr = pthread_mutex_lock(&(queue->mtx));
  if( err_thr > 0){
    perror("error lock mutex");
    retval = 1;
    pthread_exit(&retval);
  }
 
  if(opt == 0)
    {
      if( queue->nbr == queue->size) {
	err_thr = pthread_mutex_unlock(&(queue->mtx));
	
	if( err_thr > 0) {
	  perror("error unlock mutex");
	  retval = 1;
	  pthread_exit(&retval);
	}
	
	errno = EAGAIN;
	return -1;
      }
    }
  else if(opt == 2) 
    {
      // args_reset(queue);
      if(queue->in == queue->size) {
	size_t *btemp; 
	void **temp;
	
	temp = realloc(queue->buffer,
		       (size_t)(2*queue->size*sizeof(void*)));
	
	if(temp == NULL ) {
	  err_thr = pthread_mutex_unlock(&(queue->mtx));
	  
	  if( err_thr > 0){
	    perror("error unlock mutex");
	    retval = 1;
	    pthread_exit(&retval);
	  }
	    
	  errno = EAGAIN;
	  return -1;
	}
	
	btemp = realloc(queue->bufsize,
			(size_t)2*queue->size*sizeof(size_t));

	if(btemp == NULL ) {
	  err_thr = pthread_mutex_unlock(&(queue->mtx));
	  
	  if( err_thr > 0){
	    perror("error unlock mutex");
	    retval = 1;
	    pthread_exit(&retval);
	  }
	  
	  errno = EAGAIN;
	  return -1;
	}
	
	queue->buffer = temp;
	queue->bufsize = btemp;
	queue->size = queue->size*2;
      }
    }
  
  tmp  = (void*) malloc(size);
  if( tmp == NULL) {
    err_thr = pthread_mutex_unlock(&(queue->mtx));
    
    if( err_thr > 0){
      perror("error unlock mutex");
      retval = 1;
      pthread_exit(&retval);
    }
    
    errno = EAGAIN;
    return -1;
  }
  
  queue->buffer[queue->in%queue->size] = tmp;
  queue->buffer[queue->in%queue->size] = 
    memmove(queue->buffer[queue->in%queue->size],buf,size);
  
  if(queue->buffer[queue->in%queue->size] == NULL){
    perror("memmove error enqueue");
    retval = 2;
    pthread_exit(&retval);
  }
  queue->bufsize[queue->in%queue->size] = size;
  
  queue->in++;
  queue->nbr++;

  
  int *r = (int*)buf;
  printf("enqueue %d\n",(int)*r);
 

  err_thr = pthread_mutex_unlock(&(queue->mtx));
  if( err_thr > 0){
    perror("error unlock mutex");
    retval = 1;
    pthread_exit(&retval);
  }

  return 0;
}


int enqueue(struct queue *queue, const void *buf, size_t size, int blocking)
{
  /* non bloquant */
  if(blocking)
    return enqueue_block(queue,buf,size);
  /* bloquant */
  else 
    return enqueue_nblock(queue,buf,size);
}



ssize_t dequeue_block(struct queue *queue, void *buf, size_t bufsize)
{
  ssize_t lsize;
  int err_thr, retval;
  
  err_thr = pthread_mutex_lock(&(queue->mtx));
  if( err_thr > 0){
    perror("mutex lock error");
    retval = 1;
    pthread_exit(&retval);
  }

  while( queue->nbr == 0){
   err_thr = pthread_cond_wait(&(queue->empty),&(queue->mtx));
   if(err_thr > 0) 
     {
       perror("dequeue block wait cond error");
       pthread_exit(NULL);
     }
  }

  if( bufsize < queue->bufsize[queue->out%queue->size])
    {
       err_thr = pthread_mutex_unlock(&(queue->mtx));
      
       if( err_thr > 0){
	 perror("mutex unlock error");
	 retval = 1;
	 pthread_exit(&retval);
       }
       errno = EMSGSIZE;
   
       return -1;
    }
  
  lsize = queue->bufsize[queue->out%queue->size];
  
  buf = memmove(buf,queue->buffer[queue->out%queue->size],bufsize);
  
  if(buf == NULL){
    perror("memmove error dequeue");
    retval = 2;
    pthread_exit(&retval);
  }

  free(queue->buffer[queue->out%queue->size]);
  queue->out++;
  queue->nbr--;

  /*
  int *r = (int*)buf;
  printf("----> dequeue %d\n",(int)*r);
  */
  err_thr = pthread_mutex_unlock(&(queue->mtx));
  if( err_thr > 0){
    perror("mutex unlock error");
    retval = 1;
    pthread_exit(&retval);
  }
  
  err_thr = pthread_cond_signal(&(queue->full));
  if( err_thr > 0){
    perror("condition full signal error");
    retval = 1;
    pthread_exit(&retval);
  }

  return lsize;
}

ssize_t dequeue_nblock(struct queue *queue, void *buf, size_t bufsize)
{
  ssize_t lsize;
  int err_thr, retval; 
  int *r = (int*)buf;

  err_thr = pthread_mutex_lock(&queue->mtx);
  if( err_thr > 0){
    perror("mutex lock error");
    retval = 1;
    pthread_exit(&retval);
  }
  
  if( queue->nbr == 0)
    {
      err_thr = pthread_mutex_unlock(&queue->mtx);
      
      if( err_thr > 0){
	perror("mutex unlock error");
	retval = 1;
	pthread_exit(&retval);
      }
      
      errno = EAGAIN;
      return -1;
    }
  
  if( bufsize < queue->bufsize[queue->out%queue->size])
    {
      err_thr = pthread_mutex_unlock(&queue->mtx);
      
      if( err_thr > 0){
	perror("mutex unlock error");
	retval = 1;
	pthread_exit(&retval);
      }
      
      errno = EMSGSIZE;
      return -1;
    }  
 
  lsize = queue->bufsize[queue->out%queue->size]; 
  //printf("step 1 bis --> %d , %d et %d \n",queue->out,queue->size,queue->out%queue->size);
  buf = memmove(buf,queue->buffer[queue->out%queue->size],bufsize);
 
  if(buf == NULL){
    perror("memmove error dequeue");
    retval = 2;
    pthread_exit(&retval);
  }
 
  free(queue->buffer[queue->out%queue->size]);
  queue->out++;
  queue->nbr--;

  
  printf("----> dequeue %d\n",(int)*r);
  

  err_thr = pthread_mutex_unlock(&queue->mtx);
  if( err_thr > 0){
    perror("mutex unlock error");
    retval = 1;
    pthread_exit(&retval);
  }
  
  return lsize;

}

ssize_t dequeue(struct queue *queue, void *buf, size_t bufsize, int blocking)
{
  /* non bloquant */
  if(blocking)
    return dequeue_block(queue,buf,bufsize);
  /* bloquant */
  else 
    return dequeue_nblock(queue,buf,bufsize);
}


int enqueue_multiple(struct queue *queue,
                     const struct iovec *iov,
                     int iovcnt, int blocking)
{
    int i, sum, ret;
    
    if(blocking)
    {
        enqueue(queue,iov->iov_base,iov->iov_len,1);
        sum = 1;
        
        ret = enqueue_multiple(queue,(iov+1),iovcnt-1,0);
        if(ret > 0) sum+=ret;
        
    }
    else {
        sum = 0;
        for(i=0;i<iovcnt;i++){
            ret = enqueue(queue,(iov+i)->iov_base,
                          (iov+i)->iov_len,0);
            if( ret == 0) sum++;
        }
     
        if( sum == 0){
            errno = EAGAIN;
            return -1;
        } 
        
    } 
    
    return sum;
}


/* dequeue retourne des SIZE !!! De plus stocker les size successives dans les champs iov_len correspondant */
int dequeue_multiple(struct queue *queue,
                     struct iovec *iov,
                     int iovcnt, int blocking)
{
    int i, sum, ret;
    ssize_t d_size;
    
    if(blocking)
      {
        d_size = dequeue(queue,iov->iov_base,iov->iov_len,1);
        if(d_size > 0) iov->iov_len = d_size;
        
        sum = 1;
        ret = dequeue_multiple(queue,(iov+1),iovcnt-1,0);
        if(ret > 0) sum+=ret;
        
      }
    else {
      sum = 0;
      for(i=0;i<iovcnt;i++){
	d_size = dequeue(queue,(iov+i)->iov_base,(iov+i)->iov_len,0);
	if( d_size > 0){
	  sum++;
	  (iov+i)->iov_len = d_size;
	}
      }
      if( sum == 0){
	errno = EAGAIN;
	return -1;
      }
      
    }
    
    return sum;
}

/*
int main(){
  
  struct queue *queue;
  int val, i;

  queue = queue_new(8);
  i = 1;
  enqueue(queue,&i,4,0);
  i=2;
  enqueue(queue,&i,4,0);
  i=4;
  enqueue(queue,&i,4,0);
  
  dequeue(queue,&val,4,0);
  dequeue(queue,&val,4,0);
  dequeue(queue,&val,4,0);


  queue_destroy(queue);

  return 0;
}

*/

